// Copyright 2014 The mqrouter Author. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package xpulsar

import (
	"context"
	"encoding/json"
	"fmt"
	"gitee.com/go-mid/infra/xmq/mqconf"
	"strings"
	"time"

	"gitee.com/go-mid/infra/xtime"

	"github.com/apache/pulsar-client-go/pulsar"

	"gitee.com/go-mid/infra/xtrace"
)

type MessageID interface {
	pulsar.MessageID
}

type ProducerMessage struct {
	*pulsar.ProducerMessage
}

// Writer pulsar producer
type Producer struct {
	pulsar.Producer
	client pulsar.Client
	config *ProducerConfig
	url    string
}

// NewProducer return a pulsar producer
func NewProducer(ctx context.Context, producerConfig *ProducerConfig) (*Producer, error) {

	client, err := pulsar.NewClient(pulsar.ClientOptions{
		URL:              producerConfig.Broker,
		OperationTimeout: producerConfig.OperationTimeout,
	})
	if err != nil {
		return nil, fmt.Errorf("Could not instantiate Pulsar client: %v", err)
	}

	producer, err := client.CreateProducer(*producerConfig.ProducerOptions)
	if err != nil {
		client.Close()
		return nil, err
	}
	return &Producer{
		Producer: producer,
		client:   client,
		url:      strings.Join([]string{producerConfig.Broker, producerConfig.ProducerOptions.Topic}, ","),
		config:   producerConfig,
	}, nil
}

func (p *Producer) setSpanTags(span xtrace.Span) {
	span.SetTag(xtrace.TagComponent, mqconf.TraceComponent)
	span.SetTag(xtrace.TagPalfishMessageBusType, mqconf.TraceMessageBusTypePulsar)
	span.SetTag(xtrace.TagSpanKind, xtrace.SpanKindProducer)
	span.SetTag(xtrace.TagMessageBusDestination, p.config.Topic)
	span.SetTag(xtrace.TagMessagingSystem, xtrace.MessagingSystemPulsar)
	span.SetTag(xtrace.TagMessagingDestinationKind, xtrace.MessagingDestinationKindTopic)
	span.SetTag(xtrace.TagMessagingDestination, p.config.Topic)
	span.SetTag(xtrace.TagNetPeerIP, p.url)
}

// WriteMsg ...
func (p *Producer) WriteMsg(ctx context.Context, k string, v interface{}) (string, error) {
	span := xtrace.SpanFromContext(ctx)
	if span != nil {
		p.setSpanTags(span)
	}

	msg, err := json.Marshal(v)
	if err != nil {
		return "", err
	}

	st := xtime.NewTimeStat()
	messageID, err := p.Producer.Send(ctx, &pulsar.ProducerMessage{
		Payload:   msg,
		Key:       k,
		EventTime: time.Now(),
	})

	mqconf.StatReqDuration(ctx, p.Producer.Topic(), "Producer.WriteMsg", mqconf.TraceMessageBusTypePulsar, st.Millisecond())

	return string(messageID.Serialize()), err
}

func (p *Producer) Close(ctx context.Context) error {
	p.Producer.Close()
	p.client.Close()

	return nil
}
